package com.swak.reactivex.arq;

import java.util.LinkedList;
import java.util.List;
import java.util.Queue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.timer.RTO;
import com.swak.utils.Ints;
import com.swak.utils.Lists;
import com.swak.utils.Queues;
import com.swak.utils.ring.RingBuffer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.experimental.Accessors;

/**
 * ARQ.
 * 
 * @author 618lf
 */
@ToString(exclude = { "RTO_CAL", "allocator", "startTs", "flusher" })
public class ARQ {

	private static final Logger logger = LoggerFactory.getLogger(ARQ.class);

	/**
	 * MTU 的大小，需要扣除一些头部信息， Message 头部（16）+ UDP 头部（20） + IP 头部（20）
	 */
	public static final int MTU_DEF = 1400;
	public static final int ARQ_HEAD = 24;
	public static final int ACK_MIN_LEN = 8;

	/**
	 * 发送窗口刷数据时间
	 */
	public static final int RTO_NDL_MIN = 30; // 无延迟时最小的 RTO
	public static final int RTO_MIN = 100; // 延迟状态下的最小的 RTO
	public static final int RTO_DEF = 200;
	public static final int RTO_MAX = 60000;

	/**
	 * 接收窗口刷数据的时间
	 */
	public static final int INTERVAL_MIN_MS = 30;
	public static final int INTERVAL_MS = 100;
	public static final int INTERVAL_MAX_MS = 200;

	/**
	 * 默认的窗口大小
	 */
	public static final int WND = 32;

	/**
	 * 默认的缓冲区大小： 发送缓冲区 - 接收缓冲区
	 */
	public static final int SEND_BUFFER = 1024;
	public static final int RECV_BUFFER = 1024;

	/** 慢启动门限 */
	public static final int SSTHRESH_INIT = 2;
	public static final int SSTHRESH_MIN = 2;

	/**
	 * 资源释放
	 */
	private static final Recycler RECYCLER = ARQ::HandleRecycle;

	private static void HandleRecycle(Segment segment) {
		ByteBuf buf = (ByteBuf) segment.getBody();
		if (buf != null) {
			buf.release();
		}
	}

	/** 内存分配 */
	private final ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;

	/** 通过采样的 RTT 计算 RTO */
	private final RTO RTO_CAL = new RTO();

	/**
	 * 缓冲区
	 */
	private final LinkedList<Segment> sendQueue = new LinkedList<>();
	private final LinkedList<Segment> recvQueue = new LinkedList<>();
	private final SendWindow sendWindow;
	private final RecvWindow recvWindow;

	/**
	 * ACKS
	 */
	private Queue<Ack> acks = Queues.newSpscQueue();

	/** 分片控制 */
	private int mtu = MTU_DEF;
	private int mss = MTU_DEF - ARQ_HEAD;

	/** 对端实时的窗口大小 */
	private int rmtWnd = WND;

	/** 超时时间 */
	@Getter
	private int rto = RTO_DEF; // 最后一次计算的超时时间

	/** 拥塞控制 */
	private int cwnd = 1; // 拥塞窗口
	private int incr; // cwnd 的增加因子
	private int ssthresh = SSTHRESH_INIT;// cwnd 的增加阈值
	private int fastResend; // 是否快速重传，如果设置大于 0 的值则开启

	/** 刷新的时间 MS */
	@Getter
	private int interval = INTERVAL_MS;

	/** 是否开启拥流式输出 */
	private boolean streamSwitch = false;
	/** 是否开启拥塞控制 */
	private boolean congestionSwitch = true;
	/** 延迟发包 */
	private boolean nodelaySwitch = true;
	/** 延迟确认 */
	private boolean ackNodelaySwitch = false;

	/** 相对时间起始值 */
	private final long startTs = System.currentTimeMillis();

	/** 业务ID */
	private final long id;

	/** 输入输出 */
	private final Flusher flusher;

	public ARQ(long id, Flusher flusher) {
		this(id, false, flusher);
	}

	public ARQ(long id, boolean stream, Flusher flusher) {
		this(id, stream, WND, flusher);
	}

	public ARQ(long id, boolean stream, int wnd, Flusher flusher) {
		this.id = id;
		this.streamSwitch = stream;
		this.rmtWnd = wnd;
		this.flusher = flusher;
		this.sendWindow = new SendWindow(SEND_BUFFER, wnd);
		this.recvWindow = new RecvWindow(SEND_BUFFER, wnd);
	}

	/**
	 * 分配内存
	 * 
	 * @param size 内存大小
	 * @return
	 */
	public ByteBuf allotByteBuf(int size) {
		return this.allocator.ioBuffer(size);
	}

	/**
	 * 分配内存
	 * 
	 * @param size 内存大小
	 * @return
	 */
	public ByteBuf allotFlushByteBuf() {
		return this.allocator.ioBuffer(mss);
	}

	/**
	 * 当前的相对时间
	 */
	public long currentMs() {
		return System.currentTimeMillis() - startTs;
	}

	/**
	 * 先写入缓存 通过任务刷入到: 发送窗口中
	 */
	public int send(ByteBuf buf) {

		// 校验是否有数据
		int len = buf.readableBytes();
		if (len == 0) {
			return -1;
		}

		// 如果数据是 stream 类型的数据，则不需要区分什么时候结尾, 可以合并在一起.
		if (this.streamSwitch) {
			if (!sendQueue.isEmpty()) {
				Segment last = sendQueue.peekLast();
				ByteBuf lastData = (ByteBuf) last.getBody();
				int lastLen = lastData.readableBytes();
				if (lastLen < mss) {
					int cap = mss - lastLen;
					int extend = len < cap ? len : cap;
					if (lastData.maxWritableBytes() < extend) {
						ByteBuf newBuf = this.allotByteBuf(lastLen + extend);
						newBuf.writeBytes(lastData);
						lastData.release();
						lastData = newBuf;
						last.setBody(newBuf);
					}
					lastData.writeBytes(buf, extend);
				}
				last.setBodyLen(lastData.readableBytes());
			}
		}

		// 创建分片数据
		int count = len <= mss ? 1 : (len + mss - 1) / mss;
		for (int i = 0; i < count; i++) {
			int size = len > mss ? mss : len;
			Segment data = new Segment(this.id);
			data.addOption(Segment.Push);
			data.setBody(buf.readRetainedSlice(size));
			data.setBodyLen(size);
			this.sendQueue.add(data);
			if (logger.isDebugEnabled()) {
				logger.debug("创建分片:" + data);
			}
			len = buf.readableBytes();
		}

		// 写入发送窗口
		this.doSend();

		// 发送延迟
		if (this.nodelaySwitch) {
			this.flushOut();
		}

		// 返回写入成功
		return 0;
	}

	/*
	 * 写入发送窗口 : 这是一个任务需要定时的调用
	 */
	private void doSend() {

		// 当前的最多的 cwnd
		int cwnd = this.calculateCwnd();

		// 发送数据
		for (;;) {
			Segment data = this.sendQueue.peek();
			if (data == null) {
				break;
			}

			int res = this.sendWindow.send(data, cwnd);

			// 发送失败
			if (res <= 0) {
				break;
			}

			this.sendQueue.poll();

			// debug
			if (logger.isDebugEnabled()) {
				logger.debug("发送窗口:" + data);
			}
		}
	}

	public int recv(ByteBuf buf) {

		// 没有数据
		if (buf == null || buf.readableBytes() < ARQ_HEAD) {
			return -1;
		}

		// 用于记录最新的 远端的 wnd
		long lastTs = 0;

		// 当前的时间
		long currentMS = this.currentMs();

		// 是否有 Ack
		boolean hasAck = false;

		// 解析出数据
		for (;;) {

			// 解析数据
			Segment segment = null;
			try {
				segment = this.decode(buf);
			} catch (Exception e) {
				logger.error("Decode Segment Error：", e);
				break;
			}

			// 没有数据
			if (segment == null) {
				break;
			}

			if (logger.isDebugEnabled()) {
				logger.debug("接收分片：" + segment);
			}

			// 更新ACK（ack 的包才能解析出ack）
			if (segment.getAcks() != null && segment.getAcks().size() > 0) {

				// 标记有 ACK 包过来， 包括冗余的ACK
				hasAck = true;

				// 返回一个分片的真实的 rtt
				for (Ack ack : segment.getAcks()) {

					// ack
					this.sendWindow.ack(ack.getSn(), ack.getTs(), RECYCLER);

					// rtt
					long rtt = currentMS - ack.getTs();

					// rto
					RTO_CAL.calculate(rtt);
				}
			}

			// 对端同步的数据
			if (segment.isPush()) {
				this.recvQueue.add(segment);
			}

			// 设置最新的 wnd
			if (segment.getTs() > lastTs) {
				lastTs = segment.getTs();
				this.rmtWnd = segment.getWnd();
			}
		}

		// 刷数据进入： 接收窗口
		this.doRecv();

		// OLD RTO
		long oldRto = this.rto;

		// RTO
		this.rto = Ints.bound(this.nodelaySwitch ? RTO_NDL_MIN : RTO_MIN, (int) this.RTO_CAL.getTimeoutInerval(),
				RTO_MAX);

		// RTO 的变化情况也反应了 RTT 的变化情况
		boolean inr = this.rto > oldRto;

		// 开启拥塞控制
		if (this.congestionSwitch // 开启拥塞控制
				&& hasAck // 只要有 ACK 到来
				&& this.cwnd < this.rmtWnd) { // 拥塞窗口小于对端窗口
			int mss = this.mss;

			// 慢启动 阶段（快增）
			if (cwnd < ssthresh) {
				cwnd = inr ? (cwnd + 1) : (cwnd + 2);
				incr += mss;
			}

			// 拥塞避免阶段（慢增）
			else {
				if (incr < mss) {
					incr = mss;
				}
				incr += (mss * mss) / incr + (mss / 16);
				if ((cwnd + 1) * mss <= incr) {
					if (mss > 0) {
						cwnd = (incr + mss - 1) / mss;
					} else {
						cwnd = incr + mss - 1;
					}
				}
			}
			if (cwnd > rmtWnd) {
				cwnd = rmtWnd;
				incr = rmtWnd * mss;
			}
		}

		// 无延迟应答， 也会触发输入数据刷出
		if (!this.ackNodelaySwitch) {
			this.flushAcks();
			this.flushIn();
		}

		// 返回结果
		return 0;
	}

	/*
	 * 写入发送窗口
	 */
	private void doRecv() {
		for (;;) {
			Segment data = this.recvQueue.peek();
			if (data == null) {
				break;
			}

			// 收到数据
			int res = this.recvWindow.recv(data);

			// 缓存区不足
			if (res == -3) {
				break;
			}

			// 只要不是超出了缓冲区的数据，都需要接收
			// 记录 sn 和 ts； ts 能准确的计算出 rtt，从而计算出准确的 rto
			if (res != -2) {
				this.acks.add(new Ack().setSn(data.getSn()).setTs(data.getTs()));
			}

			// 只要是接收不成功的(不需要处理的)，都要释放资源
			if (res != 0) {
				RECYCLER.recycle(data);
			}

			// 弹出数据
			this.recvQueue.poll();
		}
	}

	/**
	 * 刷入数据: 任务
	 * 
	 * @return
	 */
	public int flushIn() {

		// 写入接收窗口
		this.doRecv();

		// 刷窗口数据
		int res = this.recvWindow.flush(this);

		// 没有数据 +10ms
		if (res == -1) {
			this.interval += 10;
		}
		// 有数据 -10ms
		else {
			this.interval -= 10;
		}

		// flush in interval
		this.interval = Ints.bound(INTERVAL_MIN_MS, this.interval, INTERVAL_MAX_MS);

		// 下次刷数据的时间
		return this.interval;
	}

	/**
	 * 刷出数据
	 * 
	 * @return
	 */
	public int flushOut() {

		// 准备 ack 的数据
		this.flushAcks();

		// 写入发送窗口
		this.doSend();

		// 快速重传
		int resent = this.fastResend > 0 ? this.fastResend : Integer.MAX_VALUE;

		// 当前的最多的 cwnd
		int cwnd0 = this.calculateCwnd();

		// 刷数据的 参数
		int result = this.sendWindow.flush(this.currentMs(), this.rto, cwnd0, resent, this);

		// 执行结果
		int minrto = result >> 2;
		boolean hasFastRsend = (result & 0x02) == 0x02;
		boolean hasLost = (result & 0x01) == 0x01;

		// 修改拥塞控制的数据
		if (this.congestionSwitch) {

			// 快速重传 阶段
			if (hasFastRsend) {
				int inflight = (int) (this.sendWindow.getSendNxt() - this.sendWindow.getSendUna());
				ssthresh = inflight / 2;
				if (ssthresh < SSTHRESH_MIN) {
					ssthresh = SSTHRESH_MIN;
				}
				cwnd = ssthresh + resent;
				incr = cwnd * mss;
			}

			// 超时 慢启动阶段
			if (hasLost) {
				ssthresh = this.calculateCwnd() / 2;
				if (ssthresh < SSTHRESH_MIN) {
					ssthresh = SSTHRESH_MIN;
				}
				cwnd = 1;
				incr = mss;
			}

			if (cwnd < 1) {
				cwnd = 1;
				incr = mss;
			}
		}

		return minrto;
	}

	private void flushAcks() {

		// 循环准备 ack 的数据
		for (;;) {

			if (this.acks.peek() == null) {
				break;
			}

			// buffer 写满 或者 acks 全部写完
			ByteBuf buffer = this.allotFlushByteBuf();
			this.flushAcks(buffer);

			// 创建分片
			Segment data = new Segment(this.id);
			data.addOption(Segment.Ack);
			data.setXmit(1);
			int len = buffer.readableBytes();
			data.setBody(buffer.readRetainedSlice(len));
			data.setBodyLen(len);
			data.setHeadLen(len + ARQ.ARQ_HEAD);
			buffer.release();

			// 输出数据
			this.flush2Output(data);

			// 回收 ack 的回复
			RECYCLER.recycle(data);
		}
	}

	private void flushAcks(ByteBuf buffer) {
		// 开始写入的 index.
		for (;;) {
			Ack ack = this.acks.peek();
			if (ack == null // 没有 ack 的数据
					|| buffer.writableBytes() == 0 // buffer 空间不足
					|| buffer.writableBytes() < ARQ.ACK_MIN_LEN) {// 没法写入开始数据
				break;
			}

			buffer.writeInt((int) ack.getSn());
			buffer.writeInt((int) ack.getTs());
			this.acks.poll();
		}
	}

	protected void flush2Input(Segment segment) {

		if (logger.isDebugEnabled()) {
			logger.debug("提交分片：" + segment);
		}

		try {
			ByteBuf buffer = segment.getBody();
			this.flusher.input(segment, buffer);
		} catch (Throwable e) {
			logger.error("input error", e);
		} finally {
			RECYCLER.recycle(segment);
		}

	}

	protected void flush2Output(Segment segment) {

		// 设置本地的接收窗口大小
		segment.setWnd(this.recvWindow.getCwnd());

		if (logger.isDebugEnabled()) {
			if (segment.isAck()) {
				logger.debug("确认分片:" + segment);
			} else {
				logger.debug((segment.getXmit() == 1 ? "首发" : "重传") + "分片:" + segment);
			}
		}
		ByteBuf buffer = null;
		try {
			buffer = this.encode(segment);
			this.flusher.output(segment, buffer);
		} catch (Throwable e) {
			logger.error("output error", e);

			// 如果抛出异常，则尝试释放
			if (buffer != null) {
				buffer.release();
			}
		}

		// ByteBuf 是重新分配的 Buffer， 必须在写数据之后释放
//		finally {
//			if (buffer != null) {
//				buffer.release();
//			}
//		}
	}

	/**
	 * 计算拥塞窗口
	 * 
	 * @return
	 */
	protected int calculateCwnd() {
		int cwnd0 = Ints.min(this.sendWindow.getWnd(), this.rmtWnd);
		if (this.congestionSwitch) {
			cwnd0 = Ints.min(cwnd, cwnd0);
		}
		return cwnd0;
	}

	/**
	 * for test: 不能释放ByteBuf，发送时也不能修改读写指针，因为可能需要重新发送;
	 * 
	 * @param segment
	 * @return
	 */
	public ByteBuf encode(Segment segment) {
		ByteBuf buffer = this.allotByteBuf(mtu);
		buffer.writeZero(Segment.Reserve);
		buffer.writeLong(segment.getId());
		buffer.writeInt((int) segment.getSn());
		buffer.writeInt((int) segment.getTs());
		buffer.writeByte(segment.getOption());
		buffer.writeShort(segment.getWnd());
		buffer.writeShort(segment.getHeadLen()); // 头部大小
		buffer.writeShort(segment.getBodyLen()); // 载体大小
		ByteBuf data = (ByteBuf) segment.getBody();
		if (data != null) {
			buffer.writeBytes(data, data.readerIndex(), data.readableBytes());
		}

		// 返回实际的长度
		ByteBuf slice = buffer.readRetainedSlice(buffer.readableBytes());
		buffer.release();

		// 需要写入输出 buffer 后，才能释放
		return slice;
	}

	/**
	 * for test
	 * 
	 * @param buffer
	 * @return
	 */
	public Segment decode(ByteBuf buffer) {
		if (buffer.readableBytes() < ARQ_HEAD) {
			return null;
		}
		buffer.skipBytes(Segment.Reserve);
		Segment segment = new Segment(buffer.readLong(), // id
				buffer.readInt(), // sn
				buffer.readInt(), // ts
				buffer.readByte(), // ops
				buffer.readShort(), // wnd
				buffer.readShort(), // headLen
				buffer.readShort()); // bodyLen

		// 如果有 Ack 的数据
		if (segment.isAck() && segment.getHeadLen() > ARQ_HEAD) {
			this.decodeAcks(buffer, segment);
		}

		// 返回数据 : 有数据则写入数据，否则不会写入数据
		// 读出数据后置位 ack 的数据，因为ack已经读出
		int len = segment.getBodyLen() - (segment.getHeadLen() - ARQ_HEAD);
		return (len > 0 ? segment.setBody(buffer.readRetainedSlice(len)) : segment).setHeadLen(ARQ_HEAD)
				.setBodyLen(len);
	}

	private void decodeAcks(ByteBuf buffer, Segment segment) {
		// 至少能存储一个SN
		if (segment.getHeadLen() - ARQ_HEAD >= ACK_MIN_LEN) {
			ByteBuf ackBuf = buffer.readSlice(segment.getHeadLen() - ARQ_HEAD);
			List<Ack> acks = Lists.newArrayList();
			for (; ackBuf.readableBytes() > 0;) {
				acks.add(new Ack().setSn(ackBuf.readInt()).setTs(ackBuf.readInt()));
			}
			segment.setAcks(acks);
		} else {
			buffer.skipBytes(segment.getHeadLen() - ARQ_HEAD);
		}
	}

	public void close() {
		try {
			this.flushIn();
		} catch (Exception e) {
		}

		try {
			this.flushOut();
		} catch (Exception e) {
		}

		try {
			for (;;) {
				Segment segment = this.sendQueue.poll();
				if (segment == null) {
					break;
				}
				RECYCLER.recycle(segment);
			}
		} catch (Exception e) {
		}

		try {
			for (;;) {
				Segment segment = this.recvQueue.poll();
				if (segment == null) {
					break;
				}
				RECYCLER.recycle(segment);
			}
		} catch (Exception e) {
		}

		try {
			this.sendWindow.close(RECYCLER);
		} catch (Exception e) {
		}

		try {
			this.recvWindow.close(RECYCLER);
		} catch (Exception e) {
		}
	}

	/**
	 * 回收
	 * 
	 * @author 618lf
	 */
	@FunctionalInterface
	static interface Recycler {

		/**
		 * 对 segment 进行回收处理, 主要是释放 body；
		 * 
		 * @param segment
		 */
		void recycle(Segment segment);
	}

	/**
	 * 刷数据
	 * 
	 * @author DELL
	 */
	public static interface Flusher {

		/**
		 * 输入
		 */
		void input(Segment segment, ByteBuf buffer) throws RemotingException;

		/**
		 * 输出
		 */
		void output(Segment segment, ByteBuf buffer) throws RemotingException;
	}

	/**
	 * 发送窗口
	 * 
	 * @author 618lf
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	@ToString
	static class SendWindow extends RingBuffer<Segment> {

		/**
		 * 窗口
		 */
		private int wnd;

		/**
		 * 已发送未确认序号
		 */
		private long sendUna = 0;

		/**
		 * 下一个发送的序号
		 */
		private long sendNxt = 0;

		/**
		 * SendWindow.
		 * 
		 * @param bufferSize
		 * @param wnd
		 */
		public SendWindow(int bufferSize, int wnd) {
			super(bufferSize);
			this.wnd = Ints.min(this.bufferSize, wnd);
		}

		boolean sendAble() {
			return !this.isFull(sendUna, sendNxt);
		}

		/**
		 * 发送分片数据: 统一编号
		 * 
		 * 返回值：
		 * 
		 * -1 : 不能发送数据，窗口已满 
		 * -2 : 不能发送数据，缓冲区已满 
		 * 0 : 发送数据成功 
		 * >0: 发送数据成功， 且还可以发送 N 个数据
		 * 
		 * @param data
		 * @return
		 */
		public int send(Segment data, int cwnd) {
			// sendUna + cwnd 是最大的序号
			if (this.sendNxt - this.sendUna >= cwnd) {
				return -1;
			}
			if (!this.sendAble()) {
				return -2;
			}
			data.setSn(sendNxt++);
			this.put(data.getSn(), data);

			// 返回剩余的窗口数量
			return (int) (this.wnd - (this.sendNxt - this.sendUna));
		}

		/**
		 * 刷出数据.
		 * 
		 * @param cwnd       拥塞窗口
		 * @param fastResend 快速重传的门限
		 * @return
		 */
		public int flush(long currentMS, int rto, int cwnd, int fastResend, ARQ that) {

			// 记录最小的 rto
			int minrto = RTO_MAX;
			boolean hasLost = false;
			boolean hasFastRsend = false;

			// 发送数量小于 cwnd
			int count = 0;

			// 发送的未确认的包（只统计第一次发送的次数 - 确认之后置位）
			long sn = this.sendUna;
			for (; sn < this.sendNxt && count < cwnd; sn++) {
				Segment segment = this.get(sn);
				boolean sendFalg = false;

				// 基本的校验 - 已经 Ack 的可以跳过
				if (segment == null || segment.getSn() != sn) {
					// Next
					sn++;
					continue;
				}

				// 第一次发送的数据
				if (segment.getXmit() == 0) {
					segment.setRto(rto);
					segment.setResendts(currentMS + rto);
					sendFalg = true;
				}

				// 快速重传
				else if (segment.getFastack() > fastResend) {
					segment.setFastack(0);
					segment.setRto(rto);
					segment.setResendts(currentMS + rto);
					sendFalg = true;
					hasFastRsend = true;
				}

				// 超时
				else if (currentMS > segment.getResendts()) {
					segment.setFastack(0);
					segment.setRto(Ints.add(segment.getRto(), (rto / 2)));
					segment.setResendts(currentMS + segment.getRto());
					sendFalg = true;
					hasLost = true;
				}

				// 此数据是否需要发送
				if (sendFalg) {
					segment.setXmit(Ints.add(segment.getXmit(), 1));
					segment.setTs(currentMS);
					count++;

					// 这批发送数据的最小的RTO
					if (segment.getRto() > 0 && segment.getRto() < minrto) {
						minrto = segment.getRto();
					}

					// 输出
					that.flush2Output(segment);
				}
			}

			// 返回结果
			// 最后两位是 hasFastRsend、hasLost
			int result = 0;
			result |= minrto << 2;
			result |= hasLost ? 0x01 : 0;
			result |= hasFastRsend ? 0x02 : 0;
			return result;
		}

		/**
		 * 返回值：
		 * 
		 * -1 : 不属于 [sendUna, sendNxt) 
		 *  0 : 等于 sendUna
		 *  1 : 属于 (sendUna, sendNxt)
		 * 
		 * @param ackSn
		 * @return
		 */
		public int ack(long ackSn, long ts, Recycler recycler) {

			/*
			 * check ack sn
			 */
			if (!(ackSn >= sendUna && ackSn < sendNxt)) {
				return -1;
			}

			/*
			 * 回收此缓冲区
			 */
			Segment segment = this.del(ackSn);
			if (segment != null) {
				recycler.recycle(segment);
			}

			/*
			 * 未应答边界
			 */
			if (sendUna == ackSn) {
				sendUna++;

				// 顺序推进 sendUna
				for (; sendUna < sendNxt; sendUna++) {
					Segment next = this.get(sendUna);

					// 已 ACK 的已经 DEL
					if (next != null) {
						break;
					}
				}
				return 0;
			}

			/*
			 * 统计快速发送数据
			 */
			long sn = sendUna;
			for (; sn <= ackSn - 1; sn++) {

				// 包数据
				Segment next = this.get(sn);

				// 已经确认过
				if (next == null) {
					continue;
				}

				// 当前包之前的包才能认定为需要快速重传
				if (next.getSn() != ackSn && next.getTs() <= ts) {
					next.setFastack(next.getFastack() + 1);
				}
			}

			/*
			 * 返回结果
			 */
			return 1;
		}

		void close(Recycler recycler) {
			long sn = this.sendUna;
			for (; sn < this.sendNxt; sn++) {
				Segment data = this.del(sn);
				if (data != null) {
					recycler.recycle(data);
				}
			}
		}
	}

	/**
	 * 接收窗口: 缓存的大小要大于 wnd 的大小
	 * 
	 * @author 618lf
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	@ToString
	static class RecvWindow extends RingBuffer<Segment> {

		/**
		 * 窗口
		 */
		private int wnd;

		/**
		 * 已经接收但是没有向上反馈
		 */
		private long unFlush;

		/**
		 * 期望接收的序号： 基序号(这是下一个写的序号)
		 */
		private long recvNxt = 0;

		/**
		 * RecvWindow.
		 * 
		 * @param bufferSize
		 * @param wnd
		 */
		public RecvWindow(int bufferSize, int wnd) {
			super(bufferSize);
			this.wnd = Ints.min(this.bufferSize, wnd);
		}

		boolean recvAble() {
			return !this.isFull(unFlush, recvNxt);
		}

		boolean flushAble() {
			return !this.isEmpty(unFlush, recvNxt);
		}

		/**
		 * 返回值:
		 * 
		 * -1 : 收到重复的数据 
		 * -2 : 超过窗口大小，这样的数据直接丢弃 
		 * -3 : 超过缓冲区大小 
		 * 0 : 接收成功 
		 * 1 : 重复接收
		 * 
		 * @param data
		 * @return
		 */
		public int recv(Segment data) {

			// 收到的数据的 SN
			long sn = data.getSn();

			// 这种数据需要给 ACK
			if (sn < this.recvNxt) {
				return -1;
			}

			// 超前的数据，不需要处理：（ 因为没有缓存这个数据，如果回复ack，对端以为此数据已经收到，不会在重新发送）
			// 这种数据应该直接丢弃
			if (sn > this.recvNxt + this.wnd) {
				return -2;
			}

			// 缓存区不足
			if (!this.recvAble()) {
				return -3;
			}

			/*
			 * 缓存数据： 如果是重复收到的数据，则不需要处理
			 */
			Segment old = this.put(sn, data);
			if (old != null) {
				return 1;
			}

			/*
			 * 收到基序号， 滑动窗口右边移动
			 */
			if (sn == this.recvNxt) {

				// 顺序推进 -- 不能超过边界
				long bound = this.writeBound(this.unFlush, this.recvNxt, this.wnd);
				for (; this.recvNxt < bound; this.recvNxt++) {

					// 是已经 recv 的数据
					Segment segment = this.get(this.recvNxt);
					if (segment != null) {
						// 继续下一次的循环推进
						continue;
					}

					// 此位置没有接收到的数据
					break;
				}
			}

			// 返回 确认成功
			return 0;
		}

		/**
		 * 向上传递数据
		 * 
		 * 返回值：
		 * 
		 * 0 : 没有可返回的数据 
		 * >=1 : 数据量
		 * 
		 * @return
		 */
		public int flush(ARQ that) {

			// 返回结果
			int result = -1;

			// 是否有数据需要向上传递
			if (!this.flushAble()) {
				return result;
			}

			// 刷出数据 - 向上层提交数据 - 提交之后需要回收数据
			for (; this.unFlush < this.recvNxt;) {
				Segment data = this.del(this.unFlush++);
				that.flush2Input(data);
				result++;
			}

			// 返回此次刷出的数据量
			return result;
		}

		/**
		 * 返回流量窗口: 流量窗口在可以接受的最大数据量。受到 wnd 和 实际的容量的限制；
		 * 
		 * @return
		 */
		public int getCwnd() {
			return this.calCwnd(this.unFlush, this.recvNxt, this.wnd);
		}

		void close(Recycler recycler) {
			long sn = this.unFlush;
			for (; sn < this.recvNxt; sn++) {
				Segment data = this.del(sn);
				if (data != null) {
					recycler.recycle(data);
				}
			}
		}
	}

	/**
	 * 
	 * 拥塞控制阶段
	 * 
	 * 重要的事件：
	 * A. 3个冗余的ACK，这个怎么定义 （累计确认的方式，如果是选择确认）
	 * B. 正确的ACK
	 * C. 超时
	 * 
	 * Taheo 算法： （线性增, 倍减）
	 * SS， CA， 定义一个阈值； 阈值之下是 SS 阶段，CWND 倍增; 阈值之上是 CA 阶段， CWND +1；
	 * SS、CA阶段发生 事件C ，阈值变为当前CWND的一半，CWND变为1；进入 SS 阶段；
	 * A 和 C 的处理机制一样；但是 A 发生时需要快速重传，然后进入 SS 阶段；
	 * C 发生时 超时重传，然后进入 SS 阶段；
	 * 
	 * 
	 * Reno算法：
	 * SS， CA， FR， 定义一个阈值；阈值之下是 SS 阶段，CWND 倍增; 阈值之上是 CA 阶段， CWND +1；
	 * SS、CA阶段发生事件C，阈值变为当前CWND的一半，CWND变为1；进入 SS 阶段；
	 * SS、CA阶段发生事件A，进入 FR 阶段，阈值变为当前CWND的一半，CWND变为阈值 + 3；
	 * FR 阶段 发生事件C，阈值变为当前CWND的一半，CWND变为1；进入 SS 阶段；
	 * FR 阶段 收到冗余ACK，CWND + 1；
	 * FR 阶段 收到新的ACK，CWND 变为阈值 + 3；进入 CA；
	 * 
	 * NewReno算法：
	 * 大体同 Reno算法，唯一区别：
	 * FR 进入 CA 阶段的条件是，所有段都收到ACK（收到后面ACK，重发后续的段）
	 * 
	 * @author 618lf
	 */
	static enum Phase {

		/**
		 * 慢启动
		 */
		SS,

		/**
		 * 拥塞避免
		 */
		CA,

		/**
		 * 快速恢复
		 */
		FR,

	}

	/**
	 * ACK： 最终会以 Segment 输出
	 * 
	 * @author 618lf
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	@ToString
	public static class Ack {
		/** message分片segment的序号(编码时：int) **/
		private long sn;
		/** message发送时刻的时间戳相对时间 (编码时：int)**/
		private long ts;
	}

	/**
	 * 分段： 不支持序列化
	 */
	@Getter
	@Setter
	@Accessors(chain = true)
	@ToString
	public static class Segment {

		/**
		 * 同步 Wnd
		 */
		public static final byte Sync = 1 << 1;

		/**
		 * 传递数据
		 */
		public static final byte Push = 1 << 2;

		/**
		 * 传递ACK
		 */
		public static final byte Ack = 1 << 3;

		/**
		 * 保留的 1 byte 
		 */
		public static byte Reserve = 1;

		/** message ID **/
		private long id;
		/** message分片segment的序号(编码时：int) **/
		private long sn;
		/** message发送时刻的时间戳相对时间 (编码时：int)**/
		private long ts;
		/** 选项(ack、push、sync、end(表示最后message一个包)) **/
		private byte option;
		/** 窗口大小(编码时：short) **/
		private int wnd;
		/*** 头部大小(编码时：short) 默认是24, 如果 ack， body 中 有如下长度(headLen - ARQ_HEAD)的数据是 ack;  **/
		private int headLen = ARQ_HEAD;
		/*** 数据大小(编码时：short) 最多 1400 字节 - headLen **/
		private int bodyLen;

		/** 下次超时重传的时间戳 **/
		private long resendts;
		/** 该分片的超时重传等待时间 **/
		private int rto;
		/** 收到ack时计算的该分片被跳过的累计次数，即该分片后的包都被对方收到了，达到一定次数，重传当前分片 **/
		private int fastack;
		/*** 发送方：发送分片的次数，每发送一次加一**/
		private int xmit;
		/*** 需要回复的 ack的 序号集合有序集合,此部分数据包含在body中 **/
		private List<Ack> acks;

		/*** 数据 (扩展头部 + 载体) **/
		private ByteBuf body;

		public Segment(long id) {
			this.id = id;
		}

		public Segment(long id, long sn, long ts, byte option, int wnd, int headLen, int bodyLen) {
			this.id = id;
			this.sn = sn;
			this.ts = ts;
			this.option = option;
			this.wnd = wnd;
			this.headLen = headLen;
			this.bodyLen = bodyLen;
		}

		public boolean isSync() {
			return (this.option & Sync) == Sync;
		}

		public boolean isPush() {
			return (this.option & Push) == Push;
		}

		public boolean isAck() {
			return (this.option & Ack) == Ack;
		}

		/**
		 * 添加选项
		 * 
		 * @param option
		 * @return
		 */
		public Segment addOption(byte option) {
			this.option = (byte) (this.option | option);
			return this;
		}

		/**
		 * 删除选项
		 * 
		 * @param option
		 * @return
		 */
		public Segment delOption(byte option) {
			this.option = (byte) (this.option & (~option));
			return this;
		}

		@Override
		public int hashCode() {
			return Long.hashCode(this.sn);
		}

		@Override
		public boolean equals(Object obj) {
			if (obj == null || !(obj instanceof Segment)) {
				return false;
			}
			Segment other = (Segment) obj;
			return other.getSn() == this.sn;
		}
	}
}
